package com.xuliugen.demo.storm.topology;

import com.xuliugen.demo.storm.model.Board;
import com.xuliugen.demo.storm.model.GameState;
import com.xuliugen.demo.storm.operators.ScoreFunction;
import com.xuliugen.demo.storm.operators.ScoreUpdater;
import com.xuliugen.demo.storm.operators.isEndGame;
import com.xuliugen.demo.storm.trident.spout.LocalQueueEmitter;
import com.xuliugen.demo.storm.trident.spout.LocalQueueSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.tuple.Fields;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ScoringTopology {

    private static final Logger LOG = LoggerFactory.getLogger(ScoringTopology.class);

    public static StormTopology buildTopology() {

        TridentTopology topology = new TridentTopology();

        GameState gameState = GameState.playAtRandom(new Board(), "X");
        // Scoring Queue / Spout
        LocalQueueEmitter<GameState> emitter = new LocalQueueEmitter<GameState>("ScoringQueue");
        emitter.enqueue(gameState);
        LocalQueueSpout<GameState> scoringSpout = new LocalQueueSpout<GameState>(emitter);

        Stream inputStream = topology.newStream("scoring", scoringSpout);

        inputStream.each(new Fields("gamestate"), new isEndGame())
                .each(new Fields("gamestate"), new ScoreFunction(), new Fields("board", "score", "player"))
                .each(new Fields("board", "score", "player"), new ScoreUpdater(), new Fields());
        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        final Config conf = new Config();
        final LocalCluster cluster = new LocalCluster();

        LOG.info("Submitting topology.");
        cluster.submitTopology("scoringTopology", conf, ScoringTopology.buildTopology());
        LOG.info("Topology submitted.");
        Thread.sleep(600000);
    }
}
